package nr.nodehost;

import java.io.*;
import java.nio.file.CopyOption;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardCopyOption;
import java.util.*;

import lombok.Data;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.nfunk.jep.JEP;

import net.sf.json.JSONArray;
import net.sf.json.JSONObject;
import nr.nodehost.service.AsyncTaskService;
import nr.nodehost.model.*;
import org.springframework.stereotype.Service;

@Data
@Service
@Slf4j
public class SchemaExecutor {
    public static final int MAX_LOOP_NUMBER = 100000000;

    public static final int IDLE = 0;
    public static final int PENDING = 1;
    public static final int DEPLOYING = 2;
    public static final int RUNNING = 3;
    public static final int PAUSED = 4;

    public static final String[] STATUS_INFO = {"idle", "pending", "deploying", "running", "completed"};
    private int status;
    private int progress;
    private Schema schema;
    private JSONObject executeContext;
    private Dispatcher dispatcher;
    private int loopNumber;
    @Getter
    private ArrayList<ComponentExecutor> componentExecutors = new ArrayList<ComponentExecutor>();
    private boolean needTerminate = false;
    private boolean needPause = false;

    public SchemaExecutor(Schema schema/*JSONObject schema*/, Dispatcher dispatcher) {
        this.schema = schema;//(Schema)JSONObject.toBean(schema, Schema.class);
        this.dispatcher = dispatcher;
        dispatcher.registerSchemaExecutor(this);
    }

    public boolean init(JSONObject executeContext) {
        this.executeContext = executeContext;
        schema.preprocess();
//        loopNumber = schema.getLoopNumber();
        status = PENDING;
        progress = 0;
        //new File(executeContext.getString("schemaRunDir")+"/"+schema.getId()+"/param").mkdir();
        System.out.println("schema " + schema.getId() + " inited...");
        //save params to file
        //saveParamsToFile();
        //evaluateComponentParam();
        updateStatus();     //status=1
        return true;
    }

    public int getStatus() {
        return status;
    }


    public void updateSchemaParam(JSONArray params) {
        String[] ids = schema.getComponents().keySet().toArray(new String[0]);
        for (int i = 0; i < ids.length; i++) {
            JSONObject component = schema.getComponents().get(ids[i]);
            JSONObject componentParam = component.getJSONObject("componentParam");
            String componentDir = executeContext.getString("schemaRunDir") + "/" + schema.getId() + "/" + componentParam.getString("uuid");
            ComponentDeployer.generateParameterFile(params, componentDir, componentParam);
        }
    }


    public String getSchemaRunDir() {
        return executeContext.getString("schemaRunDir") + "/" + schema.getId();
    }

    public String getComponentDir(String componentUuid) {
        return getSchemaRunDir() + "/" + componentUuid;
    }

    public String getComponentInputDir(String componentUuid) {
        return getSchemaRunDir() + "/" + componentUuid + "/input";
    }

    public String getComponentOutputDir(String componentUuid) {
        return getSchemaRunDir() + "/" + componentUuid + "/output";
    }

    public boolean updateStatus() {
        JSONObject updateResult = null;
        try {
            JSONObject statusParam = new JSONObject();

            statusParam.put("id", schema.getId());
            statusParam.put("status", status);
            statusParam.put("info", STATUS_INFO[status]);
            statusParam.put("progress", progress);

            updateResult = HttpRequestUtils.httpPost(executeContext.getString("serverBaseUrl") + "/node/updateSchemaStatus", statusParam, false);
            System.out.println("updateResult-" + updateResult);
            System.out.println("URL-" + executeContext.getString("serverBaseUrl") + "/node/updateSchemaStatus");
        } catch (Exception e) {
            e.printStackTrace();
        }
        return updateResult != null && updateResult.getInt("code") == 0;
    }

    //    public boolean terminate() {
//
//        needTerminate = true;
//
//        boolean terminated = false;
//        for (int i = 0; i < 10 && !terminated; i++) {
//            int terminatedComponentCount = 0;
//            for (ComponentExecutor ce : componentExecutors) {
//                if (ce.getStatus() == SchemaExecutor.IDLE) {
//                    terminatedComponentCount++;
//                }
//            }
//            terminated = (terminatedComponentCount == componentExecutors.size());
//            if (!terminated) {
//                try {
//                    Thread.sleep(500);
//                } catch (Exception e) {
//                    e.printStackTrace();
//                }
//            } else {
//                status = SchemaExecutor.IDLE;
//            }
//        }
//        return status == SchemaExecutor.IDLE;
//    }

    public boolean terminate_docker() {

        ArrayList<ComponentExecutor> CEList=getComponentExecutors();
        for (ComponentExecutor ce:CEList)
        {
            if(ce!=null)
            {
                String dockerName=ce.getComponentDocker();
                System.out.println("------------------------");
                System.out.println("ce : "+ ce.getComponentDocker());
                System.out.println("ID : "+ ce.getSchemaExecutor().getSchemaId());

                String flag = "docker stop "+dockerName+" && "+"docker rm "+dockerName;
                log.info("############docker stop Command："+flag);
                String[] dockerCMD = {"sh","-c",flag};
                try {
                    Process dockerStop=Runtime.getRuntime().exec(dockerCMD);
                    BufferedReader reader = new BufferedReader(new InputStreamReader(dockerStop.getInputStream()));
                    String line1;
                    while ((line1 = reader.readLine()) != null) {
                        System.out.println(line1);
                    }

                    // 读取错误输出
                    BufferedReader errorReader = new BufferedReader(new InputStreamReader(dockerStop.getErrorStream()));
                    String line;
                    while ((line = errorReader.readLine()) != null) {
                        System.out.println(line);
                    }
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
                ce.removeComponentMap();
            }
            else
            {
                return false;
            }
        }
        status = SchemaExecutor.IDLE;
        progress=0;
        updateStatus();
        return status == SchemaExecutor.IDLE;
    }
    public boolean terminate() {

        ArrayList<ComponentExecutor> CEList = getComponentExecutors();
        for (ComponentExecutor ce : CEList) {
            if (ce != null)
                if (ce.getStatus() != IDLE) {
                    Process p = ce.getComponentProcess();
                    p.destroy();
                    ce.removeComponentMap();
                } else {
                    return false;
                }

        }
        status = SchemaExecutor.IDLE;
        updateStatus();

        return status == SchemaExecutor.IDLE;
    }

    public boolean pause() {
        needPause = true;
        boolean paused = false;
        for (int i = 0; i < 10 && !paused; i++) {
            int pausedComponentCount = 0;
            for (ComponentExecutor ce : componentExecutors) {
                if (ce.getStatus() == SchemaExecutor.PAUSED) {
                    pausedComponentCount++;
                }
            }
            paused = pausedComponentCount == componentExecutors.size();
            if (!paused) {
                try {
                    Thread.sleep(500);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            } else {
                status = SchemaExecutor.PAUSED;
            }
        }
        return status == SchemaExecutor.PAUSED;
    }

    public boolean resume() {
        needPause = false;
        status = SchemaExecutor.RUNNING;
        boolean resumed = false;
        for (int i = 0; i < 10 && !resumed; i++) {
            int resumedComponentCount = 0;
            for (ComponentExecutor ce : componentExecutors) {
                if (ce.getStatus() == SchemaExecutor.RUNNING) {
                    resumedComponentCount++;
                }
            }
            resumed = (resumedComponentCount == componentExecutors.size());
            if (!resumed) {
                try {
                    Thread.sleep(100);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            } else {
                status = SchemaExecutor.RUNNING;
            }
        }
        return status == SchemaExecutor.RUNNING;
    }


    public boolean execute_docker() {
        boolean result = true;
        try {
            status = RUNNING;
            progress = 50;  //进度条
            updateStatus();
            int dataRemotePort = schema.getListenPort();
            int dataSourceSize = schema.getDataSourceSize();

            //添加docker版本
            for (JSONObject componentObj : schema.getComponents().values()) {
                JSONObject componentParam = componentObj.getJSONObject("componentParam");
                JSONObject resourceParam = componentParam.getJSONObject("resourceParam");

                //判断要部署的组件和节点对应关系
                if (!Objects.equals(resourceParam.optString("nodeIp"), executeContext.getString("nodeIp"))) {
                    continue;
                }
                //componentID=镜像名 componentUUID=容器名
                String dockerImage = componentParam.getString("id");
                String uniqueDockerName = componentParam.getString("uuid");
                //从仓库拉取镜像到本地


                //todo-多线程拉镜像
                //todo-多线程执行

                ComponentExecutor ce = new ComponentExecutor();

                // 获取相关参数值
                JSONObject functionParam = componentParam.getJSONObject("functionParam");

                int schemaId = componentParam.getInt("schemaId");
                int inputNumber = Integer.parseInt(functionParam.getString("inputNumber"));
                int dataAccess_StartStop = 0;
                AsyncTaskService taskService = NodeHostApplication.ac.getBean(AsyncTaskService.class);
                if (inputNumber == 0) {
                    System.out.println("###########dataRemotePort:"+dataRemotePort);
                    taskService.sendCmdFrame2DataAccess(schemaId, dataRemotePort, dataSourceSize, dataAccess_StartStop);
                    continue;
                }
                ce.init(dockerImage, this.executeContext, this, componentParam, dispatcher);
                componentExecutors.add(ce);
                dispatcher.registerComponentExecutor(ce);
                taskService.runComponentDocker(ce, dockerImage, uniqueDockerName);
            }
            log.info("##############这个循环出来了");
//            Thread.sleep(60*1000);      //调试使用，暂停60s
        } catch (Exception e) {
            e.printStackTrace();
            result = false;
        }
        return result;
    }
public boolean execute()
{
    boolean result = true;
    try {
        status = RUNNING;
        progress = 99;  //进度条
        updateStatus();
        int dataRemotePort = schema.getListenPort();
        int dataSourceSize = schema.getDataSourceSize();
    //Java版本
            for (JSONObject componentObj : schema.getComponents().values()) {
                JSONObject componentParam = componentObj.getJSONObject("componentParam");
                JSONObject resourceParam = componentParam.getJSONObject("resourceParam");

                if (!Objects.equals(resourceParam.optString("node"), executeContext.getString("nodeIp"))) {
                    continue;
                }
                //todo-判断要部署的容器和节点对应关系
                //todo-多线程拉镜像
                //todo-多线程执行
                //todo-前端给了完整的docker run命令
                ComponentExecutor ce = new ComponentExecutor();
                //check component input porexecuteContext.getString("t connected which type of component
                JSONObject functionParam = componentParam.getJSONObject("functionParam");
                String ComponentID = componentParam.getString("id");
                String ComponentUUID = componentParam.getString("uuid");
                int schemaId = componentParam.getInt("schemaId");
                int inputNumber = Integer.parseInt(functionParam.getString("inputNumber"));

                // 如果inputNumber为0则跳过此次循环
                // todo 12.03任务：这里的逻辑需要重写，当检测到inputNumber为0时，该组件就为DALayer，需要更改执行逻辑，进行控制帧的组帧行为
                //  假定一些参数等待传入：dataRemotePort, SrcFrameLength, 以及停止的标志位 dataAccess_StartStop
//                int dataRemotePort = 7777;
//                int dataSourceSize = 1024;
                int dataAccess_StartStop = 0;
                AsyncTaskService taskService = NodeHostApplication.ac.getBean(AsyncTaskService.class);
                if (inputNumber == 0) {
                    taskService.sendCmdFrame2DataAccess(schemaId, dataRemotePort, dataSourceSize, dataAccess_StartStop);
                    continue;
                }

/*                 // 获取组件的输入参数列表
                JSONArray inputParams = functionParam.getJSONArray("inputParams");
                for (int i = 0; i < inputNumber; i++) {
                    JSONObject inputParam = inputParams.getJSONObject(i);

                    // 获取组件的输入参数对应的源组件
                    JSONObject sourceComponent = schema.getPortSourceComponent(componentObj.getString("id"), inputParam.getString("name"));

                    // 如果源组件的类型为FILE_DATA_COMPONENT（文件类型组件），则在inputParam中添加"sourceType"字段并赋值为"file"
                    if (sourceComponent.getJSONObject("componentParam").getInt("type") == Component.FILE_DATA_COMPONENT) {
                        inputParam.put("sourceType", "file");
                    } else {
                        // 否则，在inputParam中添加"sourceType"字段并赋值为"other"
                        inputParam.put("sourceType", "other");
                    }
                }*/

                // 初始化ComponentExecutor
                ce.init(componentObj.getJSONObject("componentParam").getString("id"), this.executeContext, this, componentParam, dispatcher);

                // 将ce添加到componentExecutors列表中
                componentExecutors.add(ce);

                // 注册ce到dispatcher中
                dispatcher.registerComponentExecutor(ce);

                // 运行组件任务
                taskService.runComponent_bak(ce, ComponentID, ComponentUUID);

            }

    Thread.sleep(60*1000);      //调试使用，暂停60s
} catch (Exception e) {
        e.printStackTrace();
        result = false;
    }
        return result;
    }


    public boolean finished() {
        for (ComponentExecutor ce : componentExecutors) {
            if (!ce.finished())
                return false;
        }
        return true;
    }

    public int getExecutedNumber() {
        int number = loopNumber > 0 ? loopNumber : MAX_LOOP_NUMBER;
        for (ComponentExecutor ce : componentExecutors) {
            if (ce.getExecuteNumber() < number)
                number = ce.getExecuteNumber();
        }
        return number;
    }

    public boolean close() {
        boolean result = true;
        try {
            status = IDLE;
            progress = 0;
            updateStatus();

            //Thread.sleep(10*1000);
        } catch (Exception e) {
            result = false;
        }
        return result;
    }


    public List<JSONObject> getPortTargetComponents(String componentUuid, String portName) {

        return schema.getPortTargetComponents(componentUuid, portName);
    }

    public String getTargetPortName(String sourceComponentId, String sourceComportPort, String targetComponentId) {

        return schema.getTargetPortName(sourceComponentId, sourceComportPort, targetComponentId);
    }

    public Integer getSchemaId() {
        return schema.getId();
    }
}
